package Fp

import (
	"encoding/json"
	"fmt"
	"db2s/Er"
	"db2s/parDef"
	"math/rand"
	"strings"
	"sync/atomic"
	"time"
)

var symbol = []string{" > ", " < ", " >= ", " <= ", " = "}

func (sp TaskDistribution) rowsDataEr(object string) (t Er.RowsDataEr, err error) {
	if t, err = Er.RowsData(Er.TableRows{
		DBType: object,
	}); err != nil {
		return
	}
	return
}
func (sp TaskDistribution) MissChunkStart() (dr any, err error) {
	var (
		event = "[MissChunkStart]"
		vlog  string
	)
	if dr, err = sp.Full.GetFullReadMissST(); err != nil {
		vlog = fmt.Sprintf("(%d) %v Data query for table %v.%v without index on the source side failed!!!", sp.LogSeq, event, sp.TableObject.Schema, sp.TableObject.Table)
		errorActive(vlog)
		return
	}
	return
}
func (sp TaskDistribution) missReadSource() (sdr any, err error) {
	var (
		vlog  string
		event = "[MissChunkObject]"
	)
	vlog = fmt.Sprintf("(%d) %v Start data query of source-side non-index table %v.%v", sp.LogSeq, event, sp.TableObject.Schema, sp.TableObject.Table)
	debugActive(vlog)
	if sdr, err = sp.MissChunkStart(); err != nil {
		vlog = fmt.Sprintf("(%d) %v Data query for table %v.%v without index on the source side failed!!!", sp.LogSeq, event, sp.TableObject.Schema, sp.TableObject.Table)
		errorActive(vlog)
		return
	}
	vlog = fmt.Sprintf("(%d) %v The data query of source-side non-index table %v.%v is completed.", sp.LogSeq, event, sp.TableObject.Schema, sp.TableObject.Table)
	debugActive(vlog)
	return
}
func (sp TaskDistribution) missReadDest() (sdr any, err error) {
	var (
		vlog  string
		event = "[MissChunkObject]"
	)
	vlog = fmt.Sprintf("(%d) %v Start data query for target-side non-index table %v.%v", sp.LogSeq, event, sp.TableObject.Schema, sp.TableObject.Table)
	debugActive(vlog)
	if sdr, err = sp.MissChunkStart(); err != nil {
		vlog = fmt.Sprintf("(%d) %v Data query for table %v.%v without index on the source side failed!!!", sp.LogSeq, event, sp.TableObject.Schema, sp.TableObject.Table)
		errorActive(vlog)
		return
	}
	vlog = fmt.Sprintf("(%d) %v The data query of target-side non-index table %v.%v is completed.", sp.LogSeq, event, sp.TableObject.Schema, sp.TableObject.Table)
	debugActive(vlog)
	return
}
func (sp TaskDistribution) MissChunkObject() (dr map[string]any, err error) {
	var (
		sdr, ddr any
	)
	dr = make(map[string]any)
	switch sp.DataSource {
	case "src":
		if sdr, err = sp.missReadSource(); err != nil {
			return
		}
		dr[sp.DataSource] = sdr
	case "dst":
		if ddr, err = sp.missReadDest(); err != nil {
			return
		}
		dr[sp.DataSource] = ddr
	case "all":
		if sdr, err = sp.missReadSource(); err != nil {
			return
		}
		dr["src"] = sdr
		if ddr, err = sp.missReadDest(); err != nil {
			return
		}
		dr["dst"] = ddr
	}
	return
}
func WhereAddSplit(whereAdd string) (andGather []string) {
	var newAndGather []string
	switch {
	case strings.Contains(whereAdd, " AND "):
		newAndGather = strings.Split(whereAdd, " AND ")
	case strings.Contains(whereAdd, " and "):
		newAndGather = strings.Split(whereAdd, " and ")
	case strings.Contains(whereAdd, " OR "):
		newAndGather = strings.Split(whereAdd, " OR ")
	case strings.Contains(whereAdd, " or "):
		newAndGather = strings.Split(whereAdd, " or ")
	default:
		newAndGather = []string{whereAdd}
	}
	for _, v := range newAndGather {
		if !strings.Contains(strings.ToLower(v), "is null") || strings.Contains(strings.ToLower(v), "is not null") {
			andGather = append(andGather, v)
		}
	}
	return
}
func singleSymbol(s string, symbol string) (r string) {
	r = s
	if strings.HasPrefix(s, symbol) && strings.HasSuffix(s, symbol) {
		r = strings.TrimPrefix(r, symbol)
		r = strings.TrimSuffix(r, symbol)
	}
	return
}
func WhereAdd(v string) (symbolDirection string, r []map[string]string) {
	for _, s := range symbol {
		if strings.Contains(v, s) {
			q := strings.Split(v, s)
			switch strings.TrimSpace(s) {
			case "<", "<=":
				symbolDirection = "right"
				r = append(r, map[string]string{"columnName": strings.TrimSpace(q[0]), "columnDataRight": singleSymbol(strings.TrimSpace(q[1]), "'"), "querySymbolRight": strings.TrimSpace(s)})
			case ">", ">=":
				symbolDirection = "left"
				r = append(r, map[string]string{"columnName": strings.TrimSpace(q[0]), "columnDataLeft": singleSymbol(strings.TrimSpace(q[1]), "'"), "querySymbolLeft": strings.TrimSpace(s)})
			}
		}
	}
	return
}
func (sp TaskDistribution) prefixNullInit() {
	//处理联合索引或单列索引的首列值允许为null情况
	if len(sp.StaticP.QueryFilter) > 0 && !strings.Contains(strings.TrimSpace(strings.ToLower(sp.StaticP.QueryFilter)), " is null") { //带了条件默认排除null值
		return
	}
	if qf := ParseWhereClause(sp.StaticP.QueryFilter); len(qf) > 0 {
		var notNull = true
	bre:
		for _, v := range qf {
			_,sym,conditions := GetSingleWhereConditions(v)
			if strings.EqualFold(sym,"is") && strings.EqualFold(conditions,"null"){
				notNull = false
				break bre
			}
		}
		if notNull {
			return
		}
	}
	if func() (res bool) {
		if strings.EqualFold(getNullValue(sp.First.Null)[0], "no") {
			return true
		}
		return
	}() {
		return
	}
	var pLeft []map[string]string
	p1 := sp.pOutputInit()
	p1.WhereGenerate.IndexColumn = sp.getFirstNameStringSlice()
	swapAtomicValue(p1.WhereGenerate.LeftDirection, append(pLeft, map[string]string{"columnName": sp.getFirstNameStringSlice()[0], "columnDataLeft": "NULL", "querySymbolLeft": "="}))
	sp.firstSendMsg(*p1)
}

func (sp TaskDistribution) queryFilterInit() {
	if len(sp.StaticP.QueryFilter) == 0 {
		return
	}
	for _, v := range WhereAddSplit(sp.StaticP.QueryFilter) {
		k, j := WhereAdd(v)
		switch k {
		case "left":
			swapAtomicValue(sp.First.Batch.WhereGenerate.LeftDirection, j)
		case "right":
			swapAtomicValue(sp.First.Batch.WhereGenerate.RightDirection, j)
		}
	}
}
func (sp TaskDistribution) secondLogSeqInit() bool {
	var (
		swsStr []byte
		err    error
	)
	if swsStr, err = json.Marshal(sp.firstGetLeftDirection()); err != nil {
		return true
	}
	if m5 := md5Hash(string(swsStr)); sp.firstGetSwsMd5() != m5 {
		atomic.SwapInt64(sp.First.FirstSemLogSeq, rand.NewSource(time.Now().UnixNano()).Int63())
		return false
	}
	return true
}
func (sp TaskDistribution) protectionMechanism() {
	var (
		event = "[protectionMechanism]"
	)
	md5Str := fmt.Sprintf("table: %s.%v wherestart:%v wherecolumn:%v queryname:%v limit %v", sp.TableObject.Schema, sp.TableObject.Table, sp.firstGetWsValue(), sp.firstGetWcValue(), sp.getFirstNameStringSlice(), atomic.LoadInt64(sp.First.Batch.Limit.Pagination))
	if m5 := md5Hash(md5Str); fmt.Sprintf("%v", getAtomicValue(sp.First.BreakMd5)) == m5 {
		time.Sleep(1 * time.Second)
		swapLoopFrequency(sp.First.LoopFrequency, atomic.LoadInt64(sp.First.LoopFrequency)+1)
		if atomic.LoadInt64(sp.First.LoopFrequency) > LoopThreshold {
			errorActive(fmt.Sprintf("(%d-%v) %s setup3&: An death loop occurs in table %v.%v(%v), and the table will stop operating. md5Str info %v", sp.LogSeq, sp.getFirstSemLogSeq(), event, sp.TableObject.Schema, sp.TableObject.Table, sp.TableObject.PartitionName, md5Str))
			swapErrorPoint(sp.ErrEnd)
			return
		}
	} else {
		swapLoopFrequency(sp.First.LoopFrequency, 1)
		swapAtomicValue(sp.First.BreakMd5, m5)
	}
}

func (sp TaskDistribution) resultGroup(res1 whereLeft, res2 []whereRight) (p []parDef.IndexPart) {
	var (
		event = "[resultGroup]"
	)
	if len(res1.swsLeft) > 0 {
		swapAtomicValue(sp.First.Batch.WhereGenerate.LeftDirection, res1.swsLeft)
	}
	if len(res1.swsRight) > 0 {
		swapAtomicValue(sp.First.Batch.WhereGenerate.RightDirection, res1.swsRight)
	}
	for _, x := range res2 {
		y := sp.pOutputInit()
		y.WhereGenerate.IndexColumn = sp.getFirstNameStringSlice()
		swapAtomicValue(y.WhereGenerate.LeftDirection, x.pLeft)
		swapAtomicValue(y.WhereGenerate.RightDirection, x.pRight)
		y.PassP.Ws = sp.firstGetWsValue()
		y.PassP.Wc = sp.firstGetWcValue()
		y.LogSeq = sp.getFirstSemLogSeq()
		y.NewIndexCol = sp.getFirstNameStringSlice()
		y.Limit.Segment = x.Segment
		atomic.SwapInt64(y.Limit.Pagination, atomic.LoadInt64(sp.First.Batch.Limit.Pagination))
		infoActive(fmt.Sprintf("(%d-%v)  %v setup5&: Index segment general successful of table%v.%v(%v) limit[default] limitReset:%v general:{left:%v right:%v}.", sp.LogSeq, sp.getFirstSemLogSeq(), event, sp.TableObject.Schema, sp.TableObject.Table, sp.TableObject.PartitionName, y.Limit.Segment, sp.firstGetLeftDirection(), getAtomicValue(sp.First.Batch.WhereGenerate.RightDirection)))
		p = append(p, *y)
	}
	return
}
func (sp TaskDistribution) limitSegmentationValue() (p []parDef.IndexPart, err error) {
	var (
		compareSeq int64
		res1       whereLeft
		res2       []whereRight
	)
	if res1, res2, err = sp.conditionalValue(inputConditionalS{
		full:       sp.Full,
		Name:       sp.First.Name,
		CompareSeq: &compareSeq,
		Type:       sp.First.Type,
		Null:       sp.First.Null,
		WS:         sp.First.WS,
		WC:         sp.First.WC,
	}); err != nil {
		return
	}
	p = sp.resultGroup(res1, res2)
	switch atomic.LoadInt64(sp.First.Batch.Limit.Pagination) {
	case 0:
		switch len(sp.firstGetWsValue()) {
		case 0: //first line
		default: //end line or two line
			if atomic.LoadInt32(sp.First.Batch.Limit.ModifyS) == 0 {
				swapLimitPagination(sp.First.Batch.Limit.Pagination, *sp.StaticP.MulFactor*sp.StaticP.ChunkNumber)
			}
		}
	case 1:
		swapLimitPagination(sp.First.Batch.Limit.Pagination, 0)
	default:
		if atomic.LoadInt32(sp.First.Batch.Limit.ModifyS) == 0 {
			swapLimitPagination(sp.First.Batch.Limit.Pagination, *sp.StaticP.MulFactor*sp.StaticP.ChunkNumber)
		}
	}
	swapAtomicValue(sp.First.WS, sp.firstGetWcValue())
	return
}
func (sp TaskDistribution) outputResult() (err error) {
	var (
		event = "[First]"
		res   []parDef.IndexPart
	)
	if res, err = sp.limitSegmentationValue(); err != nil {
		return err
	}
	for _, l := range res {
		if len(getLeftDirection(l.WhereGenerate.LeftDirection)) > 0 && len(getRightDirection(l.WhereGenerate.RightDirection)) > 0 {
			infoActive(fmt.Sprintf("(%d-%v) %v setupEnd&: send Msg Index segment of table:%v.%v(%v). general sql: left %v right %v", sp.LogSeq, sp.getFirstSemLogSeq(), event, sp.TableObject.Schema, sp.TableObject.Table, sp.TableObject.PartitionName, l.WhereGenerate.LeftDirection, l.WhereGenerate.RightDirection))
			l.WhereGenerate.NewLeftDirection = getLeftDirection(l.WhereGenerate.LeftDirection)
			l.WhereGenerate.NewRightDirection = getLeftDirection(l.WhereGenerate.RightDirection)
			sp.firstSendMsg(l)
		}
	}
	return
}

// IsSegmentValue starts a query against segment
func (sp TaskDistribution) IsSegmentValue() {
	var (
		event = "[First]"
	)
	if !sp.firstRunInit() {
		sp.firstCloseMsg()
		return
	}
	sp.queryFilterInit()
	go func() {
	breakOut:
		for {
			if sp.firstErrClose(sp.LogSeq, sp.getFirstSemLogSeq()) || sp.secondLogSeqInit() {
				break breakOut
			}
			sp.First.Batch.NewIndexCol = sp.getFirstNameStringSlice()
			if err := sp.getSegmentationData(sp.First.Batch, sp.First.WC); err != nil {
				break breakOut
			}
			sp.protectionMechanism()
			vlog := fmt.Sprintf("(%d-%v) %v setup3&: Index segment query successful of table:%v.%v(%v). indexColumn:%v,sws:{left:%v right:%v ls:%v ws:%v wc:%v}", sp.LogSeq, sp.getFirstSemLogSeq(), event, sp.TableObject.Schema, sp.TableObject.Table, sp.TableObject.PartitionName, sp.First.Batch.WhereGenerate.IndexColumn, sp.firstGetLeftDirection(), sp.firstGetRightDirection(), atomic.LoadInt64(sp.First.Batch.Limit.Pagination), sp.firstGetWsValue(), sp.firstGetWcValue())
			infoActive(vlog)
			switch sp.firstNotSegmentationValue() {
			case 1:
				continue
			case 2:
				break breakOut
			}
			if sp.outputResult() != nil {
				errorActive(fmt.Sprintf("(%d-%v) %s setup3&: An death loop occurs in table %v.%v(%v), and the table will stop operating. md5Str info %v", sp.LogSeq, sp.getFirstSemLogSeq(), event, sp.TableObject.Schema, sp.TableObject.Table, sp.TableObject.PartitionName, "1"))
				swapErrorPoint(sp.ErrEnd)
				break breakOut
			}
		}
		sp.firstCloseMsg()
	}()
	return
}
